/**
 * This is the blueprint of the Coordinator objects.
 */
package com.comp6231.coordinator;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Hashtable;
import java.util.LinkedList;

import com.comp6231.common.Request;

/**
 * @author Farzana Alam
 * 
 */
public abstract class Coordinator {

	boolean isLeader = false;

	// To be used in Bully algorithm for distributed election system
	int myReplicaID = 0;

	// Port for listening operation handling messages
	public int myDataUDPPort = 0;

	// Coordinator replica name
	String myReplicaName;

	// server list - key: replica
	Hashtable<String, String[]> coordinatorReplicaList = new Hashtable<String, String[]>();
	Hashtable<String, String[]> frontEndList = new Hashtable<String, String[]>();

	private LinkedList<Request> incomingRequestQueue = new LinkedList<Request>();
	private LinkedList<String> requestProcessingOrder = new LinkedList<String>();
	private ArrayList<Request> processedRequestList = new ArrayList<Request>();
	private final Object lock = new Object();

	private String myHostName = "";

	private int myControlUDPPort;

	public Coordinator(String hostName, String dataUDPPort,
			String controlUDPPort, String replicaName, int replicaID,
			Hashtable<String, String[]> coordinatorReplicaList,
			Hashtable<String, String[]> frontEndList) {

		this.myReplicaID = replicaID;
		this.myReplicaName = replicaName;
		this.myDataUDPPort = new Integer(dataUDPPort);
		this.myHostName = hostName;
		this.myControlUDPPort = new Integer(controlUDPPort);
		this.coordinatorReplicaList = coordinatorReplicaList;
		this.frontEndList = frontEndList;

		/*
		 * The first Coordinator replica will automatically be promoted as the
		 * leader
		 */
		if (this.myReplicaID == 1) {
			formLeader(true);
		}

		// Start FIFO subsystem
		startFIFOSubsystem();

		// Start Control UDP server
		new Thread(new ControlUDPServer(Coordinator.this, myControlUDPPort))
				.start();

		// Start Data UDP server
		new Thread(new DataUDPServer(this)).start();

		// Start election subsystem
		new ElectionSubsystem(this);

		System.out.println(myReplicaName + " is Running!");
	}

	/**
	 * This method starts the FIFO subsystem
	 */
	private void startFIFOSubsystem() {
		new Thread(new FIFOSubsystem(coordinatorReplicaList, Coordinator.this,
				incomingRequestQueue)).start();
	}

	/**
	 * This method starts the Failure Detection subsystem
	 * 
	 * @return
	 */
	public boolean startFailureDetector() {
		// Start failure detection sub-system
		new FailureDetector(myReplicaName, myReplicaID, coordinatorReplicaList);
		return true;
	}

	/**
	 * Observer pattern to register Leader Coordinator
	 * 
	 * @return
	 */
	boolean notifyFrontEnd() {
		if (!isLeader)
			return false;
		return registerToFrontEnd();
	}

	/**
	 * This will create UDP message for registering the leader to be sent to the
	 * Front End
	 * 
	 * @param hostName
	 * @param messageUDPPort
	 * @return
	 */
	private boolean registerToFrontEnd() {
		String data = new Request(myHostName, myDataUDPPort, myReplicaName,
				"NA", "Register", "NA").createPacketString();
		DatagramSocket socket = null;
		String frontEndUDPPort = frontEndList.get(frontEndList.keys()
				.nextElement())[2];
		String frontEndHostName = frontEndList.get(frontEndList.keys()
				.nextElement())[0];

		try {
			socket = new DatagramSocket();
			InetAddress address = InetAddress.getByName(frontEndHostName);

			DatagramPacket request = new DatagramPacket(data.getBytes(),
					data.length(), address, new Integer(frontEndUDPPort));
			socket.send(request);
			byte[] buffer = new byte[1000];
			DatagramPacket reply = new DatagramPacket(buffer, buffer.length);
			socket.receive(reply);
			String replyMessage = new String(reply.getData()).trim();
			if (replyMessage.equalsIgnoreCase("Success"))
				return true;
			return false;
		} catch (SocketException e) {
			e.printStackTrace();
			return false;
		} catch (UnknownHostException e) {
			e.printStackTrace();
			return false;
		} catch (IOException e) {
			e.printStackTrace();
			return false;
		} finally {
			if (socket != null)
				socket.close();
		}

	}

	/**
	 * This method is used for forming a new leader after it has been elected as
	 * a leader.
	 * 
	 * @param isLeader
	 * @return
	 */
	public boolean formLeader(boolean isLeader) {
		this.isLeader = isLeader;
		if (!isLeader)
			return true;
		System.out.println(this.myReplicaName
				+ " has been elected as the new leader!");

		notifyFrontEnd();
		return true;
	}

	/**
	 * This method handles the incoming requests in UDP server.
	 * 
	 * @param UDPMessage
	 * @return
	 */
	String processRequest(Request request) {
		boolean requestProcessd = false;

		// checking if this request is already processed earlier
		if (!processedRequestList.isEmpty()
				&& processedRequestList.contains(request)) {

			return processedRequestList.get(
					processedRequestList.indexOf(request)).getResponse();
		}

		// storing the request in the FIFO queue
		synchronized (lock) {
			incomingRequestQueue.add(request);
			requestProcessingOrder.add(request.getRequestID());
			System.out.println("Request came in at " + myReplicaName + ": "
					+ request.getRequestID());
		}

		// setting the locallyProcessed flag of the Request after processing
		// locally
		request.setResponse(invokeMethod(request));

		// checking if the request was also processed in remote replicas
		while (!requestProcessd) {
			if (!requestProcessingOrder.get(0).equalsIgnoreCase(
					request.getRequestID())) {
				continue;
			}
			requestProcessd = request.requestProcessedSuccessfully();

			if (requestProcessd) {
				requestProcessingOrder.removeFirst();
			}
		}

		// storing the Request to the processedRequestList to avoid the
		// operation duplication
		processedRequestList.add(request);
		return request.getResponse();
	}

	/**
	 * This will work as an Adapter pattern for the different server
	 * implementations to work properly and sync with the distributed system.
	 * Everyones sub coordinator classes will implement this according to their
	 * respective implementations.
	 * 
	 * @param request
	 * @return
	 */
	public abstract String invokeMethod(Request request);
}
